package cn.itcast.Streams;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;

import java.util.Properties;

public class App {

    public static void main(String[] args) {
        //声明来源主题
        String fromTopic = "testStream1";
        //声明目标主题
        String toTopic = "testStream2";
        //设置参数
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"logProcessor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"ha01:9092,ha02:9092,ha03:9092");
        //实例化StreamsConfig
        StreamsConfig config = new StreamsConfig(props);
        //构建拓扑结构
        Topology topology = new Topology();
        //添加源处理节点，为源处理节点指定名称和它订阅的主题
        topology.addSource("SOURCE",fromTopic)
                //添加自定义处理节点，指定名称，处理器类和上一个节点的名称
                .addProcessor("PROCESSOR", new ProcessorSupplier() {
                    @Override
                    public Processor get() {//调用这个方法，就知道这条数据用哪个process处理，
                        return new LogProcessor();
                    }
                },"SOURCE")
                //添加目标处理节点，需要指定目标处理节点的名称，和上一个节点名称。
                .addSink("SINK",toTopic,"PROCESSOR");//最后给SINK
        //实例化KafkaStreams
        KafkaStreams streams = new KafkaStreams(topology,config);
        streams.start();
    }
}
